package cn.doitedu.rtmk.demo9;

import cn.doitedu.rtmk.common.EventBean;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class EngineProcessFunction extends KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String> {


    // 运算机池
    final HashMap<String, RuleCalculator> calculatorHashMap = new HashMap<>();

    GroovyClassLoader groovyClassLoader;

    MapState<String, RuleCalculator> calculatorStatePool;

    @Override
    public void open(Configuration parameters) throws Exception {
        groovyClassLoader = new GroovyClassLoader();

        calculatorStatePool = getRuntimeContext().getMapState(new MapStateDescriptor<String, RuleCalculator>("calculator_pool", String.class, RuleCalculator.class));

    }

    @Override
    public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {

        if(eventBean.getEventId().equals("error")  && RandomUtils.nextInt(0,5)==3){
            throw new RuntimeException("我故意异常了.......................");
        }

        /**
         * 故障恢复
         */
        // 如果运算机池中没有运算机对象，则说明系统可能被重启过，或者是新启动，或者是本来就没有运算机
        // 不管上述哪一种情况，我们都尝试从广播状态中备份的元数据，来重构运算机
        if(calculatorHashMap.isEmpty()){
            MapStateDescriptor<String, RuleMetaBean> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleMetaBean.class);
            ReadOnlyBroadcastState<String, RuleMetaBean> ruleMetaState = readOnlyContext.getBroadcastState(bcStateDesc);
            for (Map.Entry<String, RuleMetaBean> entry : ruleMetaState.immutableEntries()) {
                String ruleId = entry.getKey();
                RuleMetaBean ruleMetaBean = entry.getValue();

                // 构建并初始化运算机对象
                RuleCalculator ruleCalculator = buildCalculator(ruleMetaBean, groovyClassLoader);
                // 放入运算机池
                calculatorHashMap.put(ruleId,ruleCalculator);

                log.warn("【processElement 中恢复了一个规则运算机】,规则id:{}",ruleId);
            }
        }

        /**
         * 正常处理数据
         */
        for (Map.Entry<String, RuleCalculator> entry : calculatorHashMap.entrySet()) {
            RuleCalculator ruleCalculator = entry.getValue();
            ruleCalculator.calc(eventBean,collector);

        }

    }

    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.Context context, Collector<String> collector) throws Exception {

        /**
         * 故障恢复
         */
        if(calculatorHashMap.isEmpty()){
            MapStateDescriptor<String, RuleMetaBean> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleMetaBean.class);
            BroadcastState<String, RuleMetaBean> ruleMetaState = context.getBroadcastState(bcStateDesc);
            for (Map.Entry<String, RuleMetaBean> entry : ruleMetaState.immutableEntries()) {
                String ruleId = entry.getKey();
                RuleMetaBean ruleMetaBeanOld = entry.getValue();

                // 构建并初始化运算机对象
                RuleCalculator ruleCalculator = buildCalculator(ruleMetaBeanOld, groovyClassLoader);
                // 放入运算机池
                calculatorHashMap.put(ruleId,ruleCalculator);

                log.warn("【processBroadcastElement 中恢复了一个规则运算机】,规则id:{}",ruleId);
            }
        }


        MapStateDescriptor<String, RuleMetaBean> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleMetaBean.class);
        BroadcastState<String, RuleMetaBean> ruleMetaBroadcastState = context.getBroadcastState(bcStateDesc);

        // 收到的 ruleMetaBean ，有可能代表 管理平台 增加、修改、删除了规则参数
        // 如果操作是新增或更新后，则我们的逻辑相同，都是构造运算机，并放入运算机池
        if (("+I".equals(ruleMetaBean.getOp()) || "+U".equals(ruleMetaBean.getOp())) && ruleMetaBean.getOnlineStatus() == 1) {

            // 将元数据bean，添加或覆盖到广播状态中
            ruleMetaBroadcastState.put(ruleMetaBean.getRuleId(),ruleMetaBean);

            // 构造运行机，并使用传入的规则参数来初始化该运算机对象,现在，要从注入的代码中去构造
            RuleCalculator ruleCalculator = buildCalculator(ruleMetaBean, groovyClassLoader);

            // 将初始化好的运算机，放入运算机池
            calculatorHashMap.put(ruleMetaBean.getRuleId(),ruleCalculator);

            log.warn("收到上线规则的请求，成功上线了规则运算机: {}", ruleMetaBean.getRuleId());

        }
        // 如果操作上 删除，则我们需要把该规则从运算机池移除
        else if ("-D".equals(ruleMetaBean.getOp()) || ("+U".equals(ruleMetaBean.getOp()) && ruleMetaBean.getOnlineStatus() == 0)) {
            calculatorHashMap.remove(ruleMetaBean.getRuleId());

            // 从广播状态中删除本规则的元数据bean
            ruleMetaBroadcastState.remove(ruleMetaBean.getRuleId());

            log.warn("收到下线规则的请求，成功下线了规则运算机: {}", ruleMetaBean.getRuleId());

        }

    }


    private RuleCalculator buildCalculator(RuleMetaBean ruleMetaBean,GroovyClassLoader groovyClassLoader) throws Exception {
        // 构造运行机，并使用传入的规则参数来初始化该运算机对象,现在，要从注入的代码中去构造
        Class aClass = groovyClassLoader.parseClass(ruleMetaBean.getRuleModelCode());
        RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();


        // 初始化时，向运算机对象，注入： 规则id，规则参数json，规则的目标预圈选人群bitmap
        ruleCalculator.init(getRuntimeContext(), ruleMetaBean.getRuleParamJson(), ruleMetaBean.getPreSelectUsersBitmap());

        return ruleCalculator;
    }


}
